from decimal import localcontext
from fastapi.concurrency import run_in_threadpool
from module_stock.dao.monthly_dao import MonthlyDao
from module_stock.dao.weekly_dao import WeeklyDao
from module_stock.entity.vo.monthly_vo import MonthlyModel
from module_stock.entity.vo.weekly_vo import WeeklyModel
from utils.ak_util import fetch_akdata, filter_ak_stock
from utils.log_util import logger
from numpy import float64
from config.database import AsyncSessionLocal
from config.env import StockConfig
from module_stock.dao.basic_dao import BasicDao
from module_stock.dao.daily_dao import DailyDao
from module_stock.entity.vo.basic_vo import BasicModel
import akshare as ak

from module_stock.entity.vo.daily_vo import DailyModel
from utils.stock_util import CommonUtil, KlineUtil

async def daily_to_weekly_job(): 
    """
    股票数据从日线转成周线数据, 首先需要依赖基础数据的同步
    1.查询股票基础信息表
    2
    2.1如果没有数据，说明基础数据表未同步，提示同步基础数据表未同步的错误
    2.2如果有数据，则查询周线表所有的股票去重股票代码，并与基础信息表数据比对
    3.2.1 基础信息表有的，而周线表没有，从日线表查询日线数据，并转换为周线数据
    3.2.2 基础信息表没有，而周线表有的，从周线表进行删除
    3.2.3 都有的数据，删除当天所在周的周线表数据，查询周线表最后一条数据，根据该数据的日期
    查询日线表当天的数据，比对收盘价，如果一致，查询该日期之后的所有日线数据，转换成周线数据
    插入周线表。如果不一致，查询股票的所有数据，转换成周线数据，插入周线表。

    :param 
    :return: 
    """
    try:
        async with AsyncSessionLocal() as session:
            period='weekly'
            symbols_in_basic = await BasicDao.get_all_symbols(session)

            if symbols_in_basic and len(symbols_in_basic) > 0 :
                symbols_in_weekly = await WeeklyDao.get_all_symbols(session)
                
                #新增股票
                diff_symbols_basic = set(symbols_in_basic).difference(set(symbols_in_weekly))
                if diff_symbols_basic:                    
                    for diff_symbol_basic in diff_symbols_basic:  
                        df_daily = await DailyDao.get_all_daily_pd(session, diff_symbol_basic)
                        df_weekly = KlineUtil.convert_to_custom_freq(df_daily, 'W')
                        data_weekly = df_weekly.to_dict('records')      
                        await WeeklyDao.batch_add_weekly_dao(session, data_weekly)
                    await session.commit()
                #退市股票
                diff_symbols_weekly = set(symbols_in_weekly).difference(set(symbols_in_basic))
                if diff_symbols_weekly :
                    await WeeklyDao.batch_delete_weekly_dao(session, diff_symbols_weekly)
                    await session.commit()
                
                # 都有的股票
                same_symbols = set(symbols_in_basic).intersection(set(symbols_in_weekly))
                if same_symbols :
                    # 先删除当周的数据
                    await WeeklyDao.batch_delete_weekly_dao(session, same_symbols, CommonUtil.get_period_startdate(period))
                    for index, same_symbol in enumerate(same_symbols) :
                        last_weekly_db = await WeeklyDao.get_last_weekly_detail(session, same_symbol)
                        #因为前面统一删除了当周的数据，如果股票是当前周上市的，数据库的数据就是空的
                        if not last_weekly_db:
                            weekly_start_date = CommonUtil.get_period_startdate(period)
                        else :
                            weekly_start_date = last_weekly_db.trade_date.strftime('%Y%m%d')
                        
                        dailys_db  = await DailyDao.get_all_daily_pd_after_date(session, same_symbol, weekly_start_date)
                        dailys_db_sorted = dailys_db.sort_values(by=['trade_date'],ascending=[True])
                        
                        if last_weekly_db:
                            if last_weekly_db.close != dailys_db_sorted.iloc[0]['close'] :
                                await WeeklyDao.delete_weekly_dao(session, WeeklyModel(symbol=same_symbol))
                                dailys_db = await DailyDao.get_all_daily_pd(session, same_symbol)
                                weeklys_resample = KlineUtil.convert_to_custom_freq(dailys_db, 'W')
                                data_weekly = weeklys_resample.to_dict('records')
                                await WeeklyDao.batch_add_weekly_dao(session, data_weekly)
                            elif len(dailys_db_sorted) > 1:                          
                                data_weekly = KlineUtil.convert_to_custom_freq(dailys_db_sorted.iloc[1: , :], 'W').to_dict('records')
                                await WeeklyDao.batch_add_weekly_dao(session, data_weekly) 
                        else:
                            data_weekly = KlineUtil.convert_to_custom_freq(dailys_db_sorted, 'W').to_dict('records')
                            await WeeklyDao.batch_add_weekly_dao(session, data_weekly) 
                        if index % 10 == 9:
                            logger.info(f'从akshare同步周线数据, 执行到{index+1}次')
                            await session.commit()
            else:
                #没有数据
                raise Exception("股票基础信息表未同步")
            await session.commit()
    except Exception as e:
        await session.rollback()
        logger.exception(e)
        raise e

async def daily_to_monthly_job(): 
    """
    股票数据从日线转成月线数据, 首先需要依赖基础数据的同步
    1.查询股票基础信息表
    2
    2.1如果没有数据，说明基础数据表未同步，提示同步基础数据表未同步的错误
    2.2如果有数据，则查询月线表所有的股票去重股票代码，并与基础信息表数据比对
    3.2.1 基础信息表有的，而月线表没有，从日线表查询日线数据，并转换为月线数据
    3.2.2 基础信息表没有，而月线表有的，从月线表进行删除
    3.2.3 都有的数据，删除当天所在周的月线表数据，查询月线表最后一条数据，根据该数据的日期
    查询日线表当天的数据，比对收盘价，如果一致，查询该日期之后的所有日线数据，转换成月线数据
    插入月线表。如果不一致，查询股票的所有数据，转换成月线数据，插入月线表。

    :param 
    :return: 
    """
    try:
        async with AsyncSessionLocal()as session:
            with localcontext() as ctx:
                ctx.prec = 10
                period='monthly'
                symbols_in_basic = await BasicDao.get_all_symbols(session)

                if symbols_in_basic and len(symbols_in_basic) > 0 :
                    symbols_in_monthly = await MonthlyDao.get_all_symbols(session)
                    
                    #新增股票
                    diff_symbols_basic = set(symbols_in_basic).difference(set(symbols_in_monthly))
                    if diff_symbols_basic:                    
                        for diff_symbol_basic in diff_symbols_basic:  
                            df_daily = await DailyDao.get_all_daily_pd(session, diff_symbol_basic)
                            df_monthly = KlineUtil.convert_to_custom_freq(df_daily, 'ME')
                            data_monthly = df_monthly.to_dict('records')      
                            await MonthlyDao.batch_add_monthly_dao(session, data_monthly)
                        await session.commit()
                    #退市股票
                    diff_symbols_monthly = set(symbols_in_monthly).difference(set(symbols_in_basic))
                    if diff_symbols_monthly :
                        await MonthlyDao.batch_delete_monthly_dao(session, diff_symbols_monthly)
                        await session.commit()
                    
                    # 都有的股票
                    same_symbols = set(symbols_in_basic).intersection(set(symbols_in_monthly))
                    if same_symbols :
                        # 先删除当周的数据
                        await MonthlyDao.batch_delete_monthly_dao(session, same_symbols, CommonUtil.get_period_startdate(period))
                        for index, same_symbol in enumerate(same_symbols) :
                            last_monthly_db = await MonthlyDao.get_last_monthly_detail(session, same_symbol)
                            #因为前面统一删除了当周的数据，如果股票是当前周上市的，数据库的数据就是空的
                            if not last_monthly_db:
                                monthly_start_date = CommonUtil.get_period_startdate(period)
                            else :
                                monthly_start_date = last_monthly_db.trade_date.strftime('%Y%m%d')
                            
                            dailys_db  = await DailyDao.get_all_daily_pd_after_date(session, same_symbol, monthly_start_date)
                            dailys_db_sorted = dailys_db.sort_values(by=['trade_date'],ascending=[True])
                            
                            if last_monthly_db:
                                if last_monthly_db.close != dailys_db_sorted.iloc[0]['close'] :
                                    await MonthlyDao.delete_monthly_dao(session, MonthlyModel(symbol=same_symbol))
                                    dailys_db = await DailyDao.get_all_daily_pd(session, same_symbol)
                                    monthlys_resample = KlineUtil.convert_to_custom_freq(dailys_db, 'ME')
                                    data_monthly = monthlys_resample.to_dict('records')
                                    await MonthlyDao.batch_add_monthly_dao(session, data_monthly)
                                elif len(dailys_db_sorted) > 1:                          
                                    data_monthly = KlineUtil.convert_to_custom_freq(dailys_db_sorted.iloc[1: , :], 'ME').to_dict('records')
                                    await MonthlyDao.batch_add_monthly_dao(session, data_monthly) 
                            else:
                                data_monthly = KlineUtil.convert_to_custom_freq(dailys_db_sorted, 'ME').to_dict('records')
                                await MonthlyDao.batch_add_monthly_dao(session, data_monthly) 
                            if index % 10 == 9:
                                logger.info(f'重采月线数据, 执行到{index+1}次')
                                await session.commit()
                else:
                    #没有数据
                    raise Exception("股票基础信息表未同步")
                await session.commit()
    except Exception as e:
        await session.rollback()
        logger.exception(e)
        raise e

